package com.mj.consumer;

import com.mj.utils.Util;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * 拉模式，指定获取一个queue消息
 */
public class LitePullConsumerAssign {
    public static void main(String[] args) throws MQClientException {
        // 1.和生产者一样，创建一个DefaultMQPushConsumer实例
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(Util.ConsumerGroup);
        // 2.设置NameServer的地址，消费者通过这个地址与NameServer进行通信，来获取Broker的地址信息
        consumer.setNamesrvAddr(Util.NamesrvAddr);

        ArrayList<MessageQueue> messageQueues = new ArrayList<>(consumer.fetchMessageQueues(Util.Topic));
        consumer.assign(messageQueues);
        //指定获取offset为10的消息
        consumer.seek(messageQueues.get(0), 10);

        // 3.启动消费者实例
        consumer.start();
        System.out.println("消费者已启动");
        // 4.获取信息
        while (true) {
            List<MessageExt> messageExtList = consumer.poll();
            messageExtList.forEach(msg -> {
                System.out.println("消息：" + msg);
            });
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}
